﻿
#include <iostream>
#include <memory>
#include <thread>
#include <chrono>
#include <librdkafka/rdkafkacpp.h>

// 自定义删除器，确保安全关闭生产者
struct KafkaProducerDeleter {
    void operator()(RdKafka::Producer* producer) {
        if (producer) {
            std::cout << "安全关闭生产者..." << std::endl;

            // 1. 首先刷新所有待发送消息
            int flush_timeout = 10000; // 10秒超时
            if (producer->flush(flush_timeout) != RdKafka::ERR_NO_ERROR) {
                std::cerr << "警告: 刷新超时，可能仍有未发送消息" << std::endl;
            }

            // 2. 等待后台线程完成
            int wait_time = 3000; // 等待3秒
            while (producer->outq_len() > 0 && wait_time > 0) {
                std::cout << "等待后台线程完成，剩余消息: " << producer->outq_len() << std::endl;
                producer->poll(100);
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                wait_time -= 100;
            }

            // 3. 安全删除生产者
            delete producer;
            std::cout << "生产者已安全关闭" << std::endl;
        }
    }
};

// 日志回调类
class SafeLogger : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event& event) {
        // 立即提取信息，避免后续访问失效引用
        std::string eventType;
        switch (event.type()) {
        case RdKafka::Event::EVENT_LOG:
            eventType = "LOG";
            break;
        case RdKafka::Event::EVENT_ERROR:
            eventType = "ERROR";
            break;
        case RdKafka::Event::EVENT_STATS:
            eventType = "STATS";
            break;
        case RdKafka::Event::EVENT_THROTTLE:
            eventType = "THROTTLE";
            break;
        default:
            eventType = "UNKNOWN";
        }

        // 立即获取字符串副本，避免引用问题
        std::string message = event.str();
        std::cout << "[" << eventType << "] " << message << std::endl;
    }
};

// 发送回调类
class SafeDeliveryCallback : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message& message) {
        if (message.err()) {
            std::cerr << "消息发送失败: " << message.errstr() << std::endl;
        }
        else {
            std::cout << "消息已成功发送到分区: " << message.partition()
                << ", 偏移量: " << message.offset() << std::endl;
        }
    }
};

int main() {
    std::cout << "=== Kafka C++ 生产者测试 (安全版本) ===" << std::endl;

    std::string brokers = "127.0.0.1:9092";
    std::string topic_str = "test_topic";
    std::string errstr;

    try {
        // 1. 创建配置对象
        std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
        if (!conf) {
            throw std::runtime_error("无法创建配置对象");
        }

        // 2. 设置基本配置
        if (conf->set("bootstrap.servers", brokers, errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("设置bootstrap.servers失败: " + errstr);
        }

        // 3. 设置回调对象（确保生命周期长于生产者）
        static SafeLogger logger; // 使用静态确保生命周期
        static SafeDeliveryCallback delivery_cb;

        if (conf->set("event_cb", &logger, errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("设置event_cb失败: " + errstr);
        }

        if (conf->set("dr_cb", &delivery_cb, errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("设置dr_cb失败: " + errstr);
        }

        // 4. 创建生产者（使用智能指针和自定义删除器）
        std::unique_ptr<RdKafka::Producer, KafkaProducerDeleter> producer(
            RdKafka::Producer::create(conf.get(), errstr),
            KafkaProducerDeleter()
        );

        if (!producer) {
            throw std::runtime_error("创建生产者失败: " + errstr);
        }

        std::cout << "生产者创建成功!" << std::endl;

        // 5. 发送测试消息
        std::cout << "准备发送测试消息..." << std::endl;

        RdKafka::ErrorCode err = producer->produce(
            topic_str,
            RdKafka::Topic::PARTITION_UA,
            RdKafka::Producer::RK_MSG_COPY,
            const_cast<char*>("Hello Kafka"), 11, // 消息内容
            nullptr, 0, // 键和键长度
            0, // 时间戳
            nullptr // 消息头
        );

        if (err != RdKafka::ERR_NO_ERROR) {
            std::cerr << "发送失败: " << RdKafka::err2str(err) << std::endl;
        }
        else {
            std::cout << "消息已成功放入队列" << std::endl;
        }

        // 6. 触发发送并等待完成
        producer->poll(0);

        // 短暂等待以确保消息发送
        std::this_thread::sleep_for(std::chrono::milliseconds(2000));

        std::cout << "测试完成，准备退出..." << std::endl;

    }
    catch (const std::exception& e) {
        std::cerr << "程序异常: " << e.what() << std::endl;
        return 1;
    }

    std::cout << "=== 程序正常结束 ===" << std::endl;
    return 0;
}
//#include <iostream>
//#include <string>
//#include <librdkafka/rdkafka.h>
//
// //错误回调函数
//static void error_cb(rd_kafka_t* rk, int err, const char* reason, void* opaque) {
//    std::cerr << "错误: " << rd_kafka_err2str((rd_kafka_resp_err_t)err) << " (" << reason << ")" << std::endl;
//}
//
// //消息发送回调函数
//static void dr_msg_cb(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque) {
//    if (rkmessage->err) {
//        std::cerr << "消息发送失败: " << rd_kafka_err2str(rkmessage->err) << std::endl;
//    }
//    else {
//        std::cout << "消息发送成功，偏移量: " << rkmessage->offset << std::endl;
//    }
//}
//
// //生产者示例
//bool produce_message(const std::string& brokers, const std::string& topic, const std::string& message) {
//    rd_kafka_t* rk;          // 生产者实例
//    rd_kafka_conf_t* conf;   // 配置对象
//    char errstr[512];        // 错误信息缓冲区
//
//     //创建配置对象
//    conf = rd_kafka_conf_new();
//
//     //设置 brokers
//    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
//        std::cerr << "配置错误: " << errstr << std::endl;
//        rd_kafka_conf_destroy(conf);
//        return false;
//    }
//
//     //设置错误回调
//    rd_kafka_conf_set_error_cb(conf, error_cb);
//
//     //设置消息发送回调
//    rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);
//
//     //创建生产者实例
//    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
//    if (!rk) {
//        std::cerr << "创建生产者失败: " << errstr << std::endl;
//        return false;
//    }
//
//     //发送消息
//    rd_kafka_resp_err_t err = rd_kafka_producev(
//        rk,
//        RD_KAFKA_V_TOPIC(topic.c_str()),
//        RD_KAFKA_V_VALUE(message.c_str(), message.size()),
//        RD_KAFKA_V_OPAQUE(NULL),
//        RD_KAFKA_V_END
//    );
//
//    if (err) {
//        std::cerr << "发送消息失败: " << rd_kafka_err2str(err) << std::endl;
//        rd_kafka_destroy(rk);
//        return false;
//    }
//
//     //等待消息发送完成
//    std::cout << "等待消息发送完成..." << std::endl;
//    rd_kafka_flush(rk, 5000);  // 等待5秒
//
//     //检查是否还有未发送的消息
//    if (rd_kafka_outq_len(rk) > 0) {
//        std::cerr << "仍有 " << rd_kafka_outq_len(rk) << " 条消息未发送" << std::endl;
//    }
//
//     //清理资源
//    rd_kafka_destroy(rk);
//    return true;
//}
//
// //消费者消息处理回调
//static void consume_cb(rd_kafka_message_t* rkmessage, void* opaque) {
//    if (rkmessage->err) {
//        std::cerr << "消费错误: " << rd_kafka_err2str(rkmessage->err) << std::endl;
//        return;
//    }
//
//     //使用 API 获取主题名称，替代直接访问结构体内部成员
//    const char* topic_name = rd_kafka_topic_name(rkmessage->rkt);
//
//    std::cout << "收到消息: " << std::string((const char*)rkmessage->payload, rkmessage->len)
//        << " (主题: " << topic_name
//        << ", 分区: " << rkmessage->partition
//        << ", 偏移量: " << rkmessage->offset << ")" << std::endl;
//}
//
// //消费者示例
//bool consume_messages(const std::string& brokers, const std::string& topic, const std::string& group_id, int timeout_ms) {
//    rd_kafka_t* rk;              // 消费者实例
//    rd_kafka_conf_t* conf;       // 配置对象
//    rd_kafka_resp_err_t err;     // 错误码
//    char errstr[512];            // 错误信息缓冲区
//
//     //创建配置对象
//    conf = rd_kafka_conf_new();
//
//     //设置 brokers
//    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
//        std::cerr << "配置错误: " << errstr << std::endl;
//        rd_kafka_conf_destroy(conf);
//        return false;
//    }
//
//     //设置消费组ID
//    if (rd_kafka_conf_set(conf, "group.id", group_id.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
//        std::cerr << "配置错误: " << errstr << std::endl;
//        rd_kafka_conf_destroy(conf);
//        return false;
//    }
//
//     //设置自动提交偏移量
//    if (rd_kafka_conf_set(conf, "enable.auto.commit", "true", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
//        std::cerr << "配置错误: " << errstr << std::endl;
//        rd_kafka_conf_destroy(conf);
//        return false;
//    }
//
//     //设置错误回调
//    rd_kafka_conf_set_error_cb(conf, error_cb);
//
//     //创建消费者实例
//    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
//    if (!rk) {
//        std::cerr << "创建消费者失败: " << errstr << std::endl;
//        return false;
//    }
//
//     //订阅主题
//    rd_kafka_topic_partition_list_t* topics = rd_kafka_topic_partition_list_new(1);
//    rd_kafka_topic_partition_list_add(topics, topic.c_str(), RD_KAFKA_PARTITION_UA);  // 自动分配分区
//
//    err = rd_kafka_subscribe(rk, topics);
//    if (err) {
//        std::cerr << "订阅主题失败: " << rd_kafka_err2str(err) << std::endl;
//        rd_kafka_topic_partition_list_destroy(topics);
//        rd_kafka_destroy(rk);
//        return false;
//    }
//    rd_kafka_topic_partition_list_destroy(topics);
//
//    std::cout << "开始消费消息 (超时 " << timeout_ms << "ms)..." << std::endl;
//    std::cout << "按 Ctrl+C 停止消费" << std::endl;
//
//     //消费消息循环
//    while (true) {
//        rd_kafka_message_t* rkmessage = rd_kafka_consumer_poll(rk, timeout_ms);
//        if (!rkmessage) {
//             //超时，继续等待
//            continue;
//        }
//
//         //处理消息
//        consume_cb(rkmessage, NULL);
//
//         //释放消息
//        rd_kafka_message_destroy(rkmessage);
//    }
//
//     //取消订阅
//    rd_kafka_unsubscribe(rk);
//
//     //关闭消费者
//    rd_kafka_consumer_close(rk);
//
//     //清理资源
//    rd_kafka_destroy(rk);
//    return true;
//}
//
//int main(int argc, char** argv) {
//    // Kafka 配置 - 请根据你的实际环境修改
//    const std::string brokers = "localhost:9092";  // Kafka  broker 地址
//    const std::string topic = "test_topic";         // 测试主题名
//    const std::string group_id = "test_group";      // 消费组ID
//    const int consume_timeout = 1000;               // 消费超时时间(毫秒)
//
//    // 生产测试消息
//    std::cout << "=== 开始生产者测试 ===" << std::endl;
//    if (!produce_message(brokers, topic, "Hello, librdkafka! 这是一条测试消息")) {
//        return 1;
//    }
//
//    // 消费测试消息
//    std::cout << "\n=== 开始消费者测试 ===" << std::endl;
//    if (!consume_messages(brokers, topic, group_id, consume_timeout)) {
//        return 1;
//    }
//
//    return 0;
//}
